package com.cnlive.logpare.page

import com.cnlive.logpare.comm.CommRun
import com.cnlive.logpare.comm.CommNginxRun
import org.apache.spark.SparkConf
import java.util.Date
import org.apache.spark.SparkContext
import org.apache.spark.sql.Row
import scala.collection.mutable.HashSet
import com.cnlive.logpare.util.ParseLogUtil
import scala.collection.mutable.HashMap
import com.cnlive.logpare.util.Constant
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.LongType
import org.apache.spark.sql.types.StringType
import scala.util.control.Breaks._

object NewParsePageLogToHive extends CommRun with CommNginxRun{
    val logSign="page.js"
     val defaultAppName="NewParsePageLogToHive"
     val osList=List("nokia","sony","samsung", "htc", "sharp", "lenovo", "iphone", "ipad", "blackberry", "meizu",
            "android", "symbian", "ucweb", "opera", "cldc", "midp", "wap", "mobile", "htc", "motorola",
            "phone", "xiaomi", "oppo", "vivo", "honor", "huawei", "hisense", "zte", "playstation")
    def main(args: Array[String]): Unit = {
       val env=getEnv(args)
       val date=getDate(args)
       val files=getFilePath(env, logSign,date)
       val qqwryPath=getQqwryPath(env)//解析ip的文件地址
       val appname=getAppName(defaultAppName,date)
       val conf = new SparkConf().setAppName(appname)
       val installTime=tableNinxDateFormat.format(new Date) 
       if(env==1)conf.setMaster("local")
       val sc = new SparkContext(conf)
       val tabData= sc.textFile(files).mapPartitions(x=>{
         var result = List[Row]()
         val iPSeekerExt=getIpSeekerExt(env,qqwryPath)
         val sessionSet=new HashSet[String]
         while (x.hasNext) {
           val lineText=x.next()
           try{
               result= result.::(parLineNginxLogText(iPSeekerExt,lineText,sessionSet,parPageLogUrlandtypeToMap,mkPagelogRow,filter,installTime))
           }catch {
               case t: Throwable =>println("正行抛出异常"+lineText); t.printStackTrace() 
           }
         }
         result.iterator
       },true).filter(x=>x.getInt(0)==1).repartition(1)
        val spark=getSparkSql(env,conf)
       import spark.implicits._
       val pageLogtab=spark.createDataFrame(tabData, pageLogSchema)
       pageLogtab.registerTempTable("pageLogtab")
         val querySql2="""
             select nginxtime,uid,ip,city,location,operator,plat,isnewuser
             ,ref,url,device,browser,frmid,refdetail,urldetail,installtime,year,moth,day
               from  pageLogtab limit 10
                  """
         val querySql="""
             select max(nginxtime)
               from  pageLogtab 
                  """
       val installSql="""
             insert into TABLE  pageofflog partition(year,moth,day) 
             select nginxtime,uid,ip,city,location,operator,plat,isnewuser
             ,ref,url,device,browser,frmid,refdetail,urldetail,installtime,year,moth,day
               from  pageLogtab
               """
       if(env==1){
          spark.sql(querySql).show()
       }else{
          spark.sql("show tables").show()
          spark.sql(installSql)
       }
    }
    val filter=(lineText:String)=>{
      if(lineText.indexOf(logSign) != -1) true else false
   }
   //解析url中的数据到map中
   val  parPageLogUrlandtypeToMap=(urlandtype:List[String],mapRequest:HashMap[String, String],sessionSet:HashSet[String])=>{
       val urlandper=urlandtype(0).split(" ")
       val d=urlandper match {
              case Array(_,_,z,_*) =>{  
                  ParseLogUtil.parseUrlToMap(z, mapRequest)//解析url中的参数
                  parIsNewUser(mapRequest,sessionSet)
              }
              case _=>mapRequest.put(Constant.errCodeKey, Constant.errCodeValue_001)
       }
        var tmpbrowser=urlandtype(1).toLowerCase()
        if(tmpbrowser.indexOf("mozilla") != -1){
              tmpbrowser=tmpbrowser.substring(tmpbrowser.indexOf("mozilla")+11)
        }
         var flag=true
          breakable { 
               for(os<-osList){//判断是移动端的是PC端的
                      if(tmpbrowser.indexOf(os) != -1){
                              flag=false
                              break
                          }
                       }
               }
      if(flag){
            mapRequest.put("browser",ParseLogUtil.get_pc_browser(tmpbrowser)+"")
            mapRequest.put("plat", ParseLogUtil.get_pc_system(tmpbrowser)+"")//PC系统类型
        }else{
            mapRequest.put("device", ParseLogUtil.get_decive(tmpbrowser)+"")//手机型号
            mapRequest.put("browser", ParseLogUtil.get_mobile_browser(tmpbrowser)+"")//手机浏览器类型
            mapRequest.put("plat", ParseLogUtil.get_mobile_system(tmpbrowser)+"")//手机系统
        }
       val browser=1
   }
   //生成pageLog的一行数据
   val mkPagelogRow=(mapRequest:HashMap[String, String])=>{
          mapRequest.get(Constant.errCodeKey)  match {
                case Some(codeValue) =>if(codeValue.equals(Constant.errCodeValue_001)) Row(5)else Row(6)
                case _=>{
                       val nginxTime=tableNinxDateFormat.format(nginxDateFormat.parse(mapRequest.getOrElse("nginxtime", "19700101000000")))
                       val uid=mapRequest.getOrElse("uid", "0") 
                       val ip=mapRequest.getOrElse("ip", "127.0.0.1")
                       val city=mapRequest.getOrElse("city", "unknown")//默认值
                       val location=mapRequest.getOrElse("location", "unknown")//默认值
                       val operator=mapRequest.getOrElse("operator", "99").toInt
                       val plat=mapRequest.getOrElse("plat", "199").toInt
                       val isnewuser=mapRequest.getOrElse("isnewuser", "0").toInt//默认值
                       val installtime=mapRequest.getOrElse("installtime", "19700101000000")//默认值
                       val device=mapRequest.getOrElse("device", "199").toInt
                       val browser=mapRequest.getOrElse("browser", "0").toInt
                       val data=new HashMap[String, String]()
                       val ref=mapRequest.get("ref")  match {
                         case Some(x) =>if(x.length()>5)ParseLogUtil.playParseMap(x,data)else ""
                         case _=>""
                       }
                       val frmid=data.getOrElse("frmid", "")
                       val url=mapRequest.get("uri")  match {
                         case Some(x) =>if(x.length()>5)ParseLogUtil.playParseMap(x,data) else ""
                         case _=>""
                       }
                       val refdetail=mapRequest.getOrElse("ref", "")
                       val urldetail=mapRequest.getOrElse("uri", "")
                            //nginxtime,uid,ip,city,location,operator,plat,isnewuser,ref,url,device,browser,frmid,installtime,year,moth,day
                       Row(1,nginxTime.toLong,uid,
                           ip,city,location,operator,plat,isnewuser,ref,url,device,
                           browser,frmid,refdetail,urldetail,installtime.toLong,nginxTime.substring(0, 4),nginxTime.substring(4, 6),nginxTime.substring(6, 8)
                           )
                }
           }
    }
    //解析，看是不是一个新的用户
    def parIsNewUser(mapRequest:HashMap[String, String],sessionSet:HashSet[String]){
       if(mapRequest.contains("uid")&&mapRequest.contains("tm")){
          val uid=mapRequest.getOrElse("uid", "")
          val tm=mapRequest.getOrElse("tm", "")
          val uidAndtm=uid.split("_")
          if(uidAndtm.length==2){
            if(uidAndtm(1).length()==13&&uidAndtm(1).matches("^[0-9]*$")){
              if(uidAndtm(1).equals(tm)){
                 if(sessionSet.contains(uid)){
                    mapRequest.put("isnewuser", "0")  
                 }else{
                    sessionSet.add(uid)
                    mapRequest.put("isnewuser", "1")  
                 }
               }else{
                  mapRequest.put("isnewuser", "0")
               }
            }else if(uidAndtm(0).length()==13&&uidAndtm(0).matches("^[0-9]*$")){
              if(uidAndtm(0).equals(tm)){
                 if(sessionSet.contains(uid)){
                    mapRequest.put("isnewuser", "0")   
                 }else{
                    sessionSet.add(uid)
                    mapRequest.put("isnewuser", "1")   
                 }
               }else{
                  mapRequest.put("isnewuser", "0")
               }
            }
          }else{
             mapRequest.put("isnewuser", "0")
          }
       }else{
         mapRequest.put("isnewuser", "0")
       }
    }
     //nginxtime,uid,ip,city,location,operator,plat,isnewuser,ref,url,device,browser,frmid,installtime,year,moth,day
     //pageLog表结构
    val pageLogSchema = StructType(Seq(
        StructField("datatype", IntegerType, nullable = true),
        StructField("nginxtime", LongType, nullable = true),
        StructField("uid", StringType, nullable = true),
        StructField("ip", StringType, nullable = true),
        StructField("city", StringType, nullable = true),
        StructField("location", StringType, nullable = false),
        StructField("operator", IntegerType, nullable = false),
        StructField("plat", IntegerType, nullable = false),
        StructField("isnewuser", IntegerType, nullable = false),
        StructField("ref", StringType, nullable = false),
        StructField("url", StringType, nullable = false),
        StructField("device", IntegerType, nullable = false),
        StructField("browser", IntegerType, nullable = false),
        StructField("frmid", StringType, nullable = false),
        StructField("refdetail", StringType, nullable = false),
        StructField("urldetail", StringType, nullable = false),
        StructField("installTime", LongType, nullable = true),
        StructField("year", StringType, nullable = true),
        StructField("moth", StringType, nullable = true),
        StructField("day", StringType, nullable = true)
      ))
}